package com.lechuang.mq.config;


import com.lechuang.common.constant.BaseConstant;
import lombok.extern.log4j.Log4j2;
import org.slf4j.MDC;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;

/**
 * 消息队列配置类
 *
 * @author zyf
 */
@Configuration
@Log4j2
public class RabbitMqConfig implements RabbitListenerConfigurer {

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        //设置忽略声明异常
        rabbitAdmin.setIgnoreDeclarationExceptions(true);
        return rabbitAdmin;
    }

    @Bean
    public RabbitListenerErrorHandler rabbitListenerErrorHandler() {
        return (amqpMessage, message, exception) -> {
            exception.printStackTrace();
            throw exception;
        };
    }

    @Bean
    public MessageHandlerMethodFactory messageHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();
        messageHandlerMethodFactory.setMessageConverter(consumerJackson2MessageConverter());
        return messageHandlerMethodFactory;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
        //确认消息送到交换机(Exchange)回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            try{
                //获取correlationData中的id，也就是请求的traceId，放入mdc中，保证日志中有traceId
                String traceId = correlationData.getId();
                MDC.put(BaseConstant.TRACE_ID, traceId);
                log.info("确认消息送到交换机(Exchange)结果：");
                log.info("相关数据：" + correlationData);
                log.info("是否成功：" + ack);
                log.info("错误原因：" + cause);
            }finally {
                MDC.clear();
            }

        });

        //确认消息送到队列(Queue)回调
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
            //获取message中的traceId
            MessageProperties messageProperties = returnedMessage.getMessage().getMessageProperties();
            String traceId = (String) messageProperties.getHeaders().get(BaseConstant.TRACE_ID);
            try{
                //添加到MDC中
                MDC.put(BaseConstant.TRACE_ID, traceId);
                //获取延迟时间，如果为0，说明是即时消息，需要打印返回结果，如果大于0，说明是延迟消息，返回结果肯定是异常，所以也就没必要打印了
                Integer receivedDelay = messageProperties.getReceivedDelay();
                if(receivedDelay == 0){
                    log.info("确认消息送到队列(Queue)结果：");
                    log.info("发生消息：" + returnedMessage.getMessage());
                    log.info("回应码：" + returnedMessage.getReplyCode());
                    log.info("回应信息：" + returnedMessage.getReplyText());
                    log.info("交换机：" + returnedMessage.getExchange());
                    log.info("路由键：" + returnedMessage.getRoutingKey());
                }
            }finally {
                MDC.clear();
            }
        });
        return rabbitTemplate;
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        log.info("==== 初始化RabbitMq ==== ");
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        factory.setPrefetchCount(1);
        return factory;
    }

    @Bean
    public MappingJackson2MessageConverter consumerJackson2MessageConverter() {
        return new MappingJackson2MessageConverter();
    }

    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
        registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());
    }
}
